#ifndef _VIEW_H
#define _VIEW_H

#include "messaging.h"
#include "paxos.h"

#include <string>
#include <deque>
#include <utility>
#include <vector>
#include <pthread.h>

class view_observer {
public:
    view_observer() {}
    virtual ~view_observer() {}

    virtual void previewchange(unsigned vid) = 0;
    virtual void postviewchange(unsigned oldvid, unsigned newvid,
            const std::string &primary,
            const std::vector<std::string> &cohorts) = 0;
};

class viewmgr : public subscriber, public paxos_callback {
public:
    viewmgr(const std::string &primary, messaging *m, view_observer *obs);
    ~viewmgr();

    /**
     * List the current members in the current view including the primary.
     * If this node has not joined the replication network through the primary,
     * this function returns an emtpy vector.
     */
    std::vector<std::string> members();

    /**
     * Get the current snapshot version.
     */
//    std::string version();

    /**
     * Kick out a member. Usually called when a node is found dead.
     */
    void remove(const std::string &m);

    /**
     * Update the current snapshot version.
     * This operation can only be done at the primary.
     */
    void update_version(const std::string &newver);

    void get_view(unsigned *vid, std::string *primary,
            std::vector<std::string> *cohorts) const;

    void paxos_commit(unsigned instance, const std::string &chosen);

private:
    std::string primary_;
    std::string me_;
    std::vector<std::string> cohorts_; // always sorted
    unsigned vid_; // my current vid
    std::map<std::string, time_t> last_seen_;
    bool stable_;
    
    messaging *hermes_;
    paxos pxs_;

    std::deque<std::pair<std::string, bool> > vc_q_;

    mutable pthread_mutex_t view_m_;
    pthread_mutex_t last_seen_m_;
    pthread_cond_t stable_cv_; // if the current view has been stabilized
    pthread_cond_t recovery_cv_;
    pthread_cond_t heartbeat_cv_;
    pthread_cond_t joined_cv_;
    pthread_cond_t viewchange_cv_;
    pthread_t recovery_th_;
    pthread_t heartbeat_th_;
    pthread_t viewchange_th_;

    view_observer *vobserver_;

    /**
     * Periodically find dead peers.
     */
    void heartbeater();

    /**
     * Periodically check if this node is a member of the club. And if not,
     * contact the primary to join.
     */
    void recovery();

    void viewchange();

    bool join();

    void do_join(const std::string &peer);
    void do_remove(const std::string &peer);

    /**
     * Check if the specified node is in the current view.
     */
    bool isamember(const std::string &m,
            const std::vector<std::string> &sorted);

    void check_liveness();

    /**
     * Return the string notation of a vector of node id's.
     */
    std::string valueof(unsigned vid, const std::vector<std::string> &v);

    /**
     * Parse the string notation.
     */
    bool parse_str(const std::string &v, unsigned *vid, std::string *primary,
            std::vector<std::string> *cohorts);

    void switch_view(unsigned vid, const std::string &primary,
            const std::vector<std::string> &cohorts);

    void on_message(const DataMessage *data);

    void handle_heartbeat(const DataMessage *data);

    void handle_joinres(const DataMessage *data);

    void handle_joinreq(const DataMessage *data);

    void handle_removereq(const DataMessage *data);
};

#endif
